package com.ptone.messageCenter.api;

import com.alibaba.fastjson.JSONObject;
import com.ptone.messageCenter.common.response.ErrCode;
import com.ptone.messageCenter.common.response.Result;
import com.ptone.messageCenter.common.response.Tuple2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class ProducerAPI {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @PostMapping("/produceWithOutPartition")
    public Result produce(@RequestParam("topicName") String topicName, @RequestParam("messageKey") String messageKey, @RequestBody JSONObject jsonObject) {
        String messageValue = JSONObject.toJSONString(jsonObject);
        kafkaTemplate.send(topicName, messageKey, messageValue);
        return Result.end(Tuple2.left(ErrCode.OK));
    }

    @PostMapping("/produce")
    public Result produce(@RequestParam("topicName") String topicName, @RequestParam("partition") int partition,
                          @RequestParam("messageKey") String messageKey, @RequestParam("messageValue") String messageValue) {
        kafkaTemplate.send(topicName, partition, messageKey, messageValue);
        return Result.end(Tuple2.left(ErrCode.OK));
    }
}
